package org.fujene.communicate.repli;

import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

import org.fujene.communicate.node.Node;
import org.fujene.constants.Constants;
import org.fujene.structure.ContentInfo;
import org.fujene.structure.Queue;
import org.fujene.structure.LookupPointer;
import org.fujene.structure.ReplicateInfo;
import org.msgpack.MessagePackObject;
import org.msgpack.object.ArrayType;
import org.msgpack.object.RawType;
import org.msgpack.rpc.Client;

/**
 * Messagepack-RPC ノード間通信のためのメソッドを定義します。
 * @author iamyusk
 */
public class ReplicateCommunicator{
	private static final int PEERPORT = Constants.PEER_PORT;
	private static final int BUFFERSIZE = 100;
	private static final int TERM = Constants.TERM;
	private static final int CONTENT = Constants.CONTENT;
	private static final int ALL = Constants.ALL;

	public static ReplicateInfo getInfo(String host){
		Client client = null;
		ReplicateInfo retInfo = null;

		try{
			client = new Client(host, PEERPORT);
			ReplicatePeerHandlerInterface interfce = client.proxy(ReplicatePeerHandlerInterface.class);
			client.setRequestTimeout(3);

			// Informations
			ArrayType infoarray = interfce.getInfo();
			String[] contStr, appStr;
			byte[] graph = null;
			if(infoarray.isNil()) throw new NullPointerException("Information not packed.");
			MessagePackObject[] infos = infoarray.asArray();
			contStr = new String[infos[13].asInt()];
			appStr = new String[infos[14].asInt()];
			graph = infos[15].asByteArray();
			for(int i = 0; i < contStr.length; i++)
				contStr[i] = infos[16 + i].asString();
			for(int i = 0; i < appStr.length; i++)
				appStr[i] = infos[16 + i + contStr.length].asString();

			ContentInfo contentInfo = new ContentInfo(contStr, appStr);
			contentInfo.setConnection(graph);
			retInfo = new ReplicateInfo(infos[0].asInt(), infos[1].asInt(), infos[2].asInt(), 0,
					infos[4].asInt(), contentInfo, infos[12].asInt(), infos[5].asInt(), infos[6].asInt(),
					infos[7].asInt());
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			if(client != null) client.close();
		}

		return retInfo;
	}

	public static CopyOnWriteArrayList<Node> getNodeList(String host) throws Exception{
		CopyOnWriteArrayList<Node> nodeList = new CopyOnWriteArrayList<Node>();
		Client client = null;
		Node newNode;
		try{
			client = new Client(host, PEERPORT);
			ReplicatePeerHandlerInterface interfce = client.proxy(ReplicatePeerHandlerInterface.class);
			client.setRequestTimeout(5);

			ArrayType nodelist = interfce.nodeList();
			MessagePackObject[] objs = nodelist.asArray();
			for(int i = 0; i < objs.length; i += 2){
				newNode = Node.create(objs[i].asString(), objs[i + 1].asInt());
				nodeList.add(newNode);
			}
		}finally{
			if(client != null) client.close();
		}

		return nodeList;
	}

	public static boolean register(String host, String myAddress, int port){
		Client client = null;
		boolean result = false;
		try{
			client = new Client(host, PEERPORT);
			ReplicatePeerHandlerInterface interf = client.proxy(ReplicatePeerHandlerInterface.class);
			client.setRequestTimeout(5);

			result = interf.register(myAddress, port);
		}catch(UnknownHostException uhe){}finally{
			if(client != null) client.close();
		}
		return result;
	}

	public static ArrayType makeRemoteContentObject(Queue contents){
		MessagePackObject[] contentObjs = new MessagePackObject[contents.contentTypes
				+ contents.appendixTypes];
		for(int i = 0; i < contents.contentTypes; i++)
			contentObjs[i] = RawType.create(contents.contents[i]);
		for(int i = 0; i < contents.appendixTypes; i++)
			contentObjs[i + contents.contentTypes] = RawType.create(contents.appendices[i]);

		return ArrayType.create(contentObjs);
	}

	public static void indexContentRemote(ArrayType contents, long ID, Node node) throws Exception{
		node.setTimeout(5);
		node.worker().indexContent(contents, ID);
	}

	public static void indexTermRemote(String term, int termfreq, long ID, int type, Node node)
			throws Exception{
		node.setTimeout(5);
		node.worker().indexTerm(term, termfreq, ID, type);
	}

	public static byte[] searchRemote(String newQuery, int type, Node node) throws Exception{
		return node.worker().searchTerm(newQuery, type);
	}

	public static HashMap<String, ArrayList<LookupPointer>> getTermList(Node node, long hashFrom,
			long hashTo, String myAddress, int myPort, boolean isTermLock, boolean isNeedFuture){
		HashMap<String, ArrayList<LookupPointer>> retMap = new HashMap<String, ArrayList<LookupPointer>>();
		ArrayList<LookupPointer> ptrList = null;
		MessagePackObject[] objs = null;

		node.setTimeout(50);
		int ID = node.worker().getDataSize(TERM, hashFrom, hashTo, myAddress, myPort, isNeedFuture,
				(isTermLock ? TERM : ALL));
		System.out.println("ID: " + (ID & 0xFF));

		ArrayType datas = node.worker().getData(ID);
		objs = datas.asArray();
		if(!objs[0].isBooleanType()){
			ptrList = new ArrayList<LookupPointer>();

			for(int j = 0; 3 * j + 3 < objs.length; j++)
				ptrList.add(new LookupPointer(objs[3 * j + 2].asInt(), objs[3 * j + 3].asInt(),
						objs[3 * j + 1].asLong()));

			retMap.put(objs[0].asString(), ptrList);
		}
		if(retMap.size() % 100 == 0) System.out.printf("\rGot %8d data", retMap.size());

		System.out.printf("\rGot %8d data\nEnd of recieving data.\n", retMap.size());
		node.worker().endDataSend(ID);

		return retMap;
	}

	/**
	 * get term list.
	 * @param host ... IP address, it works as a key
	 * @param ptr ... pointer which starts reading invert index
	 * @return
	 */
	public static LinkedHashMap<Long, Integer> getInvertIndexList(Node node, LookupPointer ptr){
		LinkedHashMap<Long, Integer> invertIndices = new LinkedHashMap<Long, Integer>();

		while(!ptr.equals(LookupPointer.EMPTY)){
			// Get invert index
			ArrayType list = node.worker().getList(ptr.block, ptr.point, ptr.firstID, BUFFERSIZE);
			MessagePackObject[] hashObjs = list.asArray();
			if(hashObjs.length >= 3){
				// FIXME: Error in index
				ptr.block = hashObjs[0].asInt();
				ptr.point = hashObjs[1].asInt();
				ptr.firstID = hashObjs[2].asLong();
				for(int k = 3; k + 1 < hashObjs.length; k += 2)
					invertIndices.put(Long.valueOf(hashObjs[k].asLong()),
							Integer.valueOf(hashObjs[k + 1].asInt()));
			}
		}

		return invertIndices;
	}

	/**
	 * Get list of content via RPC.
	 */
	public static HashSet<Queue> getContentListFromNode(Node node, long hashFrom, long hashTo,
			String myAddress, int myPort, ContentInfo contentInfo, boolean isDocLock, boolean isNeedFuture){
		HashSet<Queue> retList = new HashSet<Queue>();
		int i;
		MessagePackObject[] objs;
		node.setTimeout(50);
		int dataID = node.worker().getDataSize(CONTENT, hashFrom, hashTo, myAddress, myPort, isNeedFuture,
				(isDocLock ? CONTENT : ALL));
		System.out.println("ID: " + (dataID & 0xFF));

		ArrayType datas = node.worker().getData(dataID);

		objs = datas.asArray();
		if(!objs[0].isBooleanType()){
			long ID = objs[0].asLong();
			String[] contents = new String[contentInfo.getContentNum()];
			String[] supplements = new String[contentInfo.getAppendixNum()];

			for(i = 0; i < contents.length; i++)
				contents[i] = objs[i + 1].asString();
			for(i = 0; i < supplements.length; i++)
				supplements[i] = objs[i + 1 + contents.length].asString();

			retList.add(new Queue(contents, supplements, ID));
		}
		if(retList.size() % 100 == 0) System.out.printf("\rGot %8d data", retList.size());
		System.out.printf("\rGot %8d data\nEnd of recieving data.\n", retList.size());

		node.worker().endDataSend(dataID);

		return retList;
	}

	public static Queue getContentSet(long ID, int priorIndex, int contentLength, int appendixLength,
			Node node){
		Queue retVal = null;
		int i;

		node.setTimeout(3);
		ArrayType list = node.worker().getContent(ID, priorIndex);
		MessagePackObject[] array = list.asArray();
		String[] peerContent = new String[contentLength];
		String[] peerSupplement = new String[appendixLength];

		for(i = 0; i < peerContent.length; i++)
			peerContent[i] = array[i].asString();
		for(i = 0; i < peerSupplement.length; i++)
			peerSupplement[i] = array[i + peerContent.length].asString();

		retVal = new Queue(peerContent, peerSupplement, ID);

		return retVal;
	}

	public static boolean isLive(Node node) throws Exception{
		boolean retb = false;
		node.setTimeout(3);
		if(node.worker().isLive() != null) retb = true;

		return retb;
	}

	public static void finalizeProcess(Node remoteNode, Node ownNode){
		try{
			remoteNode.worker().deleteNode(ownNode.getAddr(), ownNode.getPort());
		}finally{
			if(remoteNode.client() != null) remoteNode.closeConnection();
		}
	}

	public static void deleteContent(Node node, long ID){
		node.worker().deleteContent(ID);
	}

	public static void synchronizeClock(Node node){
		node.worker().synchronizeClock();
	}

	public static void stopIndexing(Node node){
		node.worker().stopIndexing();
	}

	public static void resumeIndexing(Node node){
		node.worker().resumeIndexing();
	}

	public static void backupLatest(Node node){
		node.worker().backupLatest();
	}

	public static void revertLatest(Node node){
		node.worker().revertLatest();
	}

	public static void suspend(Node node, Node ownNode, long suspendTime){
		node.worker().suspend(ownNode.getAddr(), ownNode.getPort(), suspendTime);
	}
}
